package com.learn.kafka.assign;


import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 向指定分区发送消息的生产者（继承通用配置父类）
 * @author: learn.kafka
 */
public class ProducerSendTest extends ProducerProperties {


    public static void main(String[] args) {
        new ProducerSendTest().sendToSpecificPartitions();
    }

    // 构造方法：可根据需要覆盖父类配置
    public ProducerSendTest() {
        super(); // 调用父类构造方法，加载通用配置
        // 如需修改配置，可在此处覆盖
        // 例如：setConfig("bootstrap.servers", "192.168.44.160:9092");
    }

    // 向指定分区发送消息的逻辑
    private void sendToSpecificPartitions() {
        // 从父类获取生产者实例
        Producer<Integer, String> producer = createProducer();

        try {
            // 向主题 "ass5part" 的不同分区发送消息
            producer.send(new ProducerRecord<>("ass5part", 0, 0, "0"));
            producer.send(new ProducerRecord<>("ass5part", 1, 1, "1"));
            producer.send(new ProducerRecord<>("ass5part", 2, 2, "2"));
            producer.send(new ProducerRecord<>("ass5part", 3, 3, "3"));
            producer.send(new ProducerRecord<>("ass5part", 4, 4, "4"));

            System.out.println("消息已成功发送到指定分区");
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}
